---
title: task_engine
sidebarTitle: task_engine
---

# `prefect.task_engine`

## Functions

### `run_task_sync` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1512" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
run_task_sync(task: 'Task[P, R]', task_run_id: Optional[UUID] = None, task_run: Optional[TaskRun] = None, parameters: Optional[dict[str, Any]] = None, wait_for: Optional['OneOrManyFutureOrResult[Any]'] = None, return_type: Literal['state', 'result'] = 'result', dependencies: Optional[dict[str, set[RunInput]]] = None, context: Optional[dict[str, Any]] = None) -> Union[R, State, None]
```

### `run_task_async` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1543" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
run_task_async(task: 'Task[P, R]', task_run_id: Optional[UUID] = None, task_run: Optional[TaskRun] = None, parameters: Optional[dict[str, Any]] = None, wait_for: Optional['OneOrManyFutureOrResult[Any]'] = None, return_type: Literal['state', 'result'] = 'result', dependencies: Optional[dict[str, set[RunInput]]] = None, context: Optional[dict[str, Any]] = None) -> Union[R, State, None]
```

### `run_generator_task_sync` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1574" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
run_generator_task_sync(task: 'Task[P, R]', task_run_id: Optional[UUID] = None, task_run: Optional[TaskRun] = None, parameters: Optional[dict[str, Any]] = None, wait_for: Optional['OneOrManyFutureOrResult[Any]'] = None, return_type: Literal['state', 'result'] = 'result', dependencies: Optional[dict[str, set[RunInput]]] = None, context: Optional[dict[str, Any]] = None) -> Generator[R, None, None]
```

### `run_generator_task_async` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1633" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
run_generator_task_async(task: 'Task[P, R]', task_run_id: Optional[UUID] = None, task_run: Optional[TaskRun] = None, parameters: Optional[dict[str, Any]] = None, wait_for: Optional['OneOrManyFutureOrResult[Any]'] = None, return_type: Literal['state', 'result'] = 'result', dependencies: Optional[dict[str, set[RunInput]]] = None, context: Optional[dict[str, Any]] = None) -> AsyncGenerator[R, None]
```

### `run_task` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1719" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
run_task(task: 'Task[P, Union[R, Coroutine[Any, Any, R]]]', task_run_id: Optional[UUID] = None, task_run: Optional[TaskRun] = None, parameters: Optional[dict[str, Any]] = None, wait_for: Optional['OneOrManyFutureOrResult[Any]'] = None, return_type: Literal['state', 'result'] = 'result', dependencies: Optional[dict[str, set[RunInput]]] = None, context: Optional[dict[str, Any]] = None) -> Union[R, State, None, Coroutine[Any, Any, Union[R, State, None]]]
```


Runs the provided task.

**Args:**
- `task`: The task to run
- `task_run_id`: The ID of the task run; if not provided, a new task run
will be created
- `task_run`: The task run object; if not provided, a new task run
will be created
- `parameters`: The parameters to pass to the task
- `wait_for`: A list of futures to wait for before running the task
- `return_type`: The return type to return; either "state" or "result"
- `dependencies`: A dictionary of task run inputs to use for dependency tracking
- `context`: A dictionary containing the context to use for the task run; only
required if the task is running on in a remote environment

**Returns:**
- The result of the task run


## Classes

### `TaskRunTimeoutError` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L115" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>


Raised when a task run exceeds its timeout.


### `BaseTaskRunEngine` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L120" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

**Methods:**

#### `compute_transaction_key` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L157" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
compute_transaction_key(self) -> Optional[str]
```

#### `handle_rollback` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L297" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
handle_rollback(self, txn: Transaction) -> None
```

#### `is_cancelled` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L148" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
is_cancelled(self) -> bool
```

#### `is_running` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L253" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
is_running(self) -> bool
```

Whether or not the engine is currently running a task.


#### `log_finished_message` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L259" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
log_finished_message(self) -> None
```

#### `record_terminal_state_timing` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L246" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
record_terminal_state_timing(self, state: State) -> None
```

#### `state` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L143" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
state(self) -> State
```

### `SyncTaskRunEngine` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L314" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

**Methods:**

#### `asset_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L685" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
asset_context(self)
```

#### `begin_run` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L398" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
begin_run(self) -> None
```

#### `call_hooks` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L361" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
call_hooks(self, state: Optional[State] = None) -> None
```

#### `call_task_fn` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L890" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
call_task_fn(self, transaction: Transaction) -> Union[ResultRecord[Any], None, Coroutine[Any, Any, R], R]
```

Convenience method to call the task function. Returns a coroutine if the
task is async.


#### `can_retry` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L324" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
can_retry(self, exc_or_state: Exception | State[R]) -> bool
```

#### `client` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L319" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
client(self) -> SyncPrefectClient
```

#### `handle_crash` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L631" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
handle_crash(self, exc: BaseException) -> None
```

#### `handle_exception` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L599" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
handle_exception(self, exc: Exception) -> None
```

#### `handle_retry` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L544" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
handle_retry(self, exc_or_state: Exception | State[R]) -> bool
```

Handle any task run retries.

- If the task has retries left, and the retry condition is met, set the task to retrying and return True.
- If the task has a retry delay, place in AwaitingRetry state with a delayed scheduled time.
- If the task has no retries left, or the retry condition is not met, return False.


#### `handle_success` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L504" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
handle_success(self, result: R, transaction: Transaction) -> Union[ResultRecord[R], None, Coroutine[Any, Any, R], R]
```

#### `handle_timeout` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L616" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
handle_timeout(self, exc: TimeoutError) -> None
```

#### `initialize_run` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L703" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
initialize_run(self, task_run_id: Optional[UUID] = None, dependencies: Optional[dict[str, set[RunInput]]] = None) -> Generator[Self, Any, Any]
```

Enters a client context and creates a task run if needed.


#### `result` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L489" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
result(self, raise_on_failure: bool = True) -> 'Union[R, State, None]'
```

#### `run_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L870" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
run_context(self)
```

#### `set_state` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L438" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
set_state(self, state: State[R], force: bool = False) -> State[R]
```

#### `setup_run_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L641" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
setup_run_context(self, client: Optional[SyncPrefectClient] = None)
```

#### `start` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L800" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
start(self, task_run_id: Optional[UUID] = None, dependencies: Optional[dict[str, set[RunInput]]] = None) -> Generator[None, None, None]
```

#### `transaction_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L843" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
transaction_context(self) -> Generator[Transaction, None, None]
```

#### `wait_until_ready` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L777" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
wait_until_ready(self) -> None
```

Waits until the scheduled time (if its the future), then enters Running.


### `AsyncTaskRunEngine` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L907" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

**Methods:**

#### `asset_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1292" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
asset_context(self)
```

#### `begin_run` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L990" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
begin_run(self) -> None
```

#### `call_hooks` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L953" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
call_hooks(self, state: Optional[State] = None) -> None
```

#### `call_task_fn` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1496" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
call_task_fn(self, transaction: AsyncTransaction) -> Union[ResultRecord[Any], None, Coroutine[Any, Any, R], R]
```

Convenience method to call the task function. Returns a coroutine if the
task is async.


#### `can_retry` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L917" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
can_retry(self, exc_or_state: Exception | State[R]) -> bool
```

#### `client` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L912" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
client(self) -> PrefectClient
```

#### `handle_crash` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1237" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
handle_crash(self, exc: BaseException) -> None
```

#### `handle_exception` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1205" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
handle_exception(self, exc: Exception) -> None
```

#### `handle_retry` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1149" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
handle_retry(self, exc_or_state: Exception | State[R]) -> bool
```

Handle any task run retries.

- If the task has retries left, and the retry condition is met, set the task to retrying and return True.
- If the task has a retry delay, place in AwaitingRetry state with a delayed scheduled time.
- If the task has no retries left, or the retry condition is not met, return False.


#### `handle_success` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1112" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
handle_success(self, result: R, transaction: AsyncTransaction) -> Union[ResultRecord[R], None, Coroutine[Any, Any, R], R]
```

#### `handle_timeout` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1220" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
handle_timeout(self, exc: TimeoutError) -> None
```

#### `initialize_run` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1310" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
initialize_run(self, task_run_id: Optional[UUID] = None, dependencies: Optional[dict[str, set[RunInput]]] = None) -> AsyncGenerator[Self, Any]
```

Enters a client context and creates a task run if needed.


#### `result` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1097" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
result(self, raise_on_failure: bool = True) -> 'Union[R, State, None]'
```

#### `run_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1476" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
run_context(self)
```

#### `set_state` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1044" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
set_state(self, state: State, force: bool = False) -> State
```

#### `setup_run_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1248" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
setup_run_context(self, client: Optional[PrefectClient] = None)
```

#### `start` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1405" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
start(self, task_run_id: Optional[UUID] = None, dependencies: Optional[dict[str, set[RunInput]]] = None) -> AsyncGenerator[None, None]
```

#### `transaction_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1450" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
transaction_context(self) -> AsyncGenerator[AsyncTransaction, None]
```

#### `wait_until_ready` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1382" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
wait_until_ready(self) -> None
```

Waits until the scheduled time (if its the future), then enters Running.

